#include "stdafx.h"
#include "msg/notificationqueue.hpp"
#include "msg/notificationcenter.hpp"
#include "msg/notification.hpp"
#include "designpattern/singletonholder.hpp"
using namespace JHCPP::designpattern;
#include "stdlib/bugcheck.hpp"
using namespace JHCPP::stdlib;

namespace JHCPP
{
	namespace msg
	{
		CNotificationQueue::CNotificationQueue()
		{
		}

		CNotificationQueue::~CNotificationQueue()
		{
			try
			{
				clear();
			}
			catch (...)
			{
				jh_unexpected();
			}
		}

		void CNotificationQueue::enqueueNotification(CNotification::Ptr pNotification)
		{
			jh_check_ptr (pNotification);
			CMutex::ScopedLock lock(m_mutex);
			if (m_waitQueue.empty())
			{
				m_nfQueue.push_back(pNotification);
			}
			else
			{
				WaitInfo* pWI = m_waitQueue.front();
				m_waitQueue.pop_front();
				pWI->pNf = pNotification;
				pWI->nfAvailable.set();
			}	
		}

		void CNotificationQueue::enqueueUrgentNotification(CNotification::Ptr pNotification)
		{
			jh_check_ptr (pNotification);
			CMutex::ScopedLock lock(m_mutex);
			if (m_waitQueue.empty())
			{
				m_nfQueue.push_front(pNotification);
			}
			else
			{
				WaitInfo* pWI = m_waitQueue.front();
				m_waitQueue.pop_front();
				pWI->pNf = pNotification;
				pWI->nfAvailable.set();
			}	
		}

		CNotification* CNotificationQueue::dequeueNotification()
		{
			CMutex::ScopedLock lock(m_mutex);
			return dequeueOne().add();
		}

		CNotification* CNotificationQueue::waitDequeueNotification()
		{
			CNotification::Ptr pNf;
			WaitInfo* pWI = 0;
			{
				CMutex::ScopedLock lock(m_mutex);
				pNf = dequeueOne();
				if (pNf) return pNf.add();
				pWI = new WaitInfo;
				m_waitQueue.push_back(pWI);
			}
			pWI->nfAvailable.wait();
			pNf = pWI->pNf;
			delete pWI;
			return pNf.add();
		}

		CNotification* CNotificationQueue::waitDequeueNotification(long milliseconds)
		{
			CNotification::Ptr pNf;
			WaitInfo* pWI = 0;
			{
				CMutex::ScopedLock lock(m_mutex);
				pNf = dequeueOne();
				if (pNf) return pNf.add();
				pWI = new WaitInfo;
				m_waitQueue.push_back(pWI);
			}
			if (pWI->nfAvailable.tryWait(milliseconds))
			{
				pNf = pWI->pNf;
			}
			else
			{
				CMutex::ScopedLock lock(m_mutex);
				pNf = pWI->pNf;
				for (WaitQueue::iterator it = m_waitQueue.begin(); it != m_waitQueue.end(); ++it)
				{
					if (*it == pWI)
					{
						m_waitQueue.erase(it);
						break;
					}
				}
			}
			delete pWI;
			return pNf.add();
		}

		void CNotificationQueue::dispatch(CNotificationCenter& notificationCenter)
		{
			CMutex::ScopedLock lock(m_mutex);
			CNotification::Ptr pNf = dequeueOne();
			while (pNf)
			{
				notificationCenter.postNotification(pNf);
				pNf = dequeueOne();
			}
		}

		void CNotificationQueue::wakeUpAll()
		{
			CMutex::ScopedLock lock(m_mutex);
			for (WaitQueue::iterator it = m_waitQueue.begin(); it != m_waitQueue.end(); ++it)
			{
				(*it)->nfAvailable.set();
			}
			m_waitQueue.clear();
		}

		bool CNotificationQueue::empty() const
		{
			CMutex::ScopedLock lock(m_mutex);
			return m_nfQueue.empty();
		}

		int CNotificationQueue::size() const
		{
			CMutex::ScopedLock lock(m_mutex);
			return static_cast<int>(m_nfQueue.size());
		}

		void CNotificationQueue::clear()
		{
			CMutex::ScopedLock lock(m_mutex);
			m_nfQueue.clear();	
		}

		bool CNotificationQueue::hasIdleThreads() const
		{
			CMutex::ScopedLock lock(m_mutex);
			return !m_waitQueue.empty();
		}

		CNotification::Ptr CNotificationQueue::dequeueOne()
		{
			CNotification::Ptr pNf;
			if (!m_nfQueue.empty())
			{
				pNf = m_nfQueue.front();
				m_nfQueue.pop_front();
			}
			return pNf;
		}

		static TSingletonHolder<CNotificationQueue> sh;
		CNotificationQueue& CNotificationQueue::defaultQueue()
		{
			return *sh.get();
		}

	}//end of namespace JHCPP
}//end of namespace msg
